package bancosys.tec.job.dispatcher;

import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import jmine.tec.batch.BatchController;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import bancosys.tec.job.Job;
import bancosys.tec.job.JobDispatcher;
import bancosys.tec.job.JobDispatchingException;
import bancosys.tec.job.JobExecutionException;

/**
 * Dispatcher que simula filas JMS
 * 
 * @created May 26, 2006
 * @author Gustavo Almeida
 */
public class ThreadJobDispatcher implements JobDispatcher {

    private static final int DEFAULT_MAX_THREADS = 5;

    private static Log log = LogFactory.getLog(ThreadJobDispatcher.class);

    private int threadNumber = 0;

    private BatchController batchController;

    private int maxRunningThreads = DEFAULT_MAX_THREADS;

    private Set<Thread> runningThreads = new HashSet<Thread>(this.maxRunningThreads);

    private List<Thread> waitingThreads = new LinkedList<Thread>();

    private Object notified;

    /**
     * Devolve o número máximo de threads que pode estar rodando concorrentemente.
     * 
     * @return o número máximo de threads que pode estar rodando concorrentemente.
     */
    public int getMaxRunningThreads() {
        return this.maxRunningThreads;
    }

    /**
     * Define o número máximo de threads que pode estar rodando concorrentemente.
     * 
     * @param numberOfThreads o número máximo de threads que pode estar rodando concorrentemente.
     */
    public void setNumberOfThreads(int numberOfThreads) {
        this.maxRunningThreads = numberOfThreads;
    }

    /**
     * Define o objeto que deve ser notificado quando a fila de threads estiver vazia.
     * 
     * @param notified o objeto que deve ser notificado quando a fila de threads estiver vazia.
     * @see Object#notify()
     * @see Object#wait()
     */
    public void setNotified(Object notified) {
        this.notified = notified;
    }

    /**
     * {@inheritDoc}
     */
    public void dispatch(Job job) throws JobDispatchingException {
        Thread thread = new JobThread(job);

        synchronized (this.runningThreads) {
            this.waitingThreads.add(thread);
            log.debug("thread waiting: " + thread + ", r:" + this.runningThreads.size() + "/w:" + this.waitingThreads.size());
            this.startThreads();
        }
    }

    /**
     * É chamado para cada thread que é finalizada, notifica <code>notified</code> quando as filas ficam vazias.
     * 
     * @param thread thread finalida.
     */
    private void threadFinished(Thread thread) {
        synchronized (this.runningThreads) {
            this.runningThreads.remove(thread);
            log.debug("thread finished: " + thread + ", r:" + this.runningThreads.size() + "/w:" + this.waitingThreads.size());
            this.startThreads();
            if (this.runningThreads.isEmpty() && this.notified != null) {
                synchronized (this.notified) {
                    this.notified.notify();
                }
            }
        }
    }

    /**
     * Devolve a quantidade de threads rodando somada com a quantidade de threads que estão esperando para serem executadas.
     * 
     * @return a quantidade total de threads nas filas.
     */
    public int getTotalThreads() {
        return this.runningThreads.size() + this.waitingThreads.size();
    }

    /**
     * Inicia as threads que estão na fila de espera se o numero de threads em execução permitir.
     */
    private void startThreads() {
        synchronized (this.runningThreads) {
            while (this.runningThreads.size() < this.maxRunningThreads && this.waitingThreads.size() > 0) {
                Iterator<Thread> it = this.waitingThreads.iterator();
                Thread thread = it.next();
                it.remove();
                this.runningThreads.add(thread);
                thread.start();
                log.debug("thread started: " + thread + ", r:" + this.runningThreads.size() + "/w:" + this.waitingThreads.size());
            }
        }
    }

    /**
     * Thread que encapsula o Job.
     * 
     * @author Gustavo Almeida
     */
    public class JobThread extends Thread {
        private Job job;

        /**
         * Construtor.
         * 
         * @param job job.
         */
        public JobThread(Job job) {
            this.job = job;
            this.setDaemon(false);
            this.setName("JobThread" + ThreadJobDispatcher.this.threadNumber++);
        }

        /**
         * Executa o job.
         */
        @Override
        public void run() {
            try {
                ThreadJobDispatcher.this.batchController.clearBatchExecutionContext();
                this.job.execute(ThreadJobDispatcher.this.batchController.getBatchExecutionContext());
            } catch (JobExecutionException e) {
                log.error("Job[" + this.job.getId() + "]: execution failed.", e);
            } finally {
                ThreadJobDispatcher.this.threadFinished(this);
            }
        }

    }

    /**
     * @return the batchController
     */
    public BatchController getBatchController() {
        return this.batchController;
    }

    /**
     * @param batchController the batchController to set
     */
    public void setBatchController(BatchController batchController) {
        this.batchController = batchController;
    }

}
